package com.jianying.exer;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * ClassName: Window_EventTimeSliding_MaxVc
 * Package: com.atguigu.exer
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/10 15:08
 * @Version: 1.0
 */
public class Window_EventTimeSliding_MaxVc {
    public static void main(String[] args) throws Exception {
        // TODO 在事件时间语义下,编码滑动时间窗口计算每个传感器的最大vc.
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> wsDS = env.socketTextStream("localhost", 8888)
                .map(new WaterSensorMapFunction());

        SingleOutputStreamOperator<WaterSensor> withWMDS = wsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        // .<WaterSensor>forBoundedOutOfOrderness(Duration.ofMillis(3))
                        .<WaterSensor>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs() ;
                            }
                        })
        );



        withWMDS.keyBy(ws -> ws.getId())
                .window(SlidingEventTimeWindows.of(Time.milliseconds(10), Time.milliseconds(5)))
                        .maxBy("vc")
                                .print();


        env.execute();
    }
}
